package pvt_chat

import (
	"context"
	"encoding/json"
	"errors"
	"github.com/go-redis/redis"
	"github.com/olivere/elastic/v7"
	global2 "longmen/server/config/global"
	"longmen/server/pkg/common/lib/bytesconv"
	"strconv"
	"sync"
	"time"
)

// OnlineDeviceRequest 在线设备
type OnlineDeviceRequest struct {
	DeviceId       string `json:"device_id"`        // 设备ID
	Platform       int    `json:"platform"`         // 设备平台
	UserID         string `json:"user_id"`          // 用户ID
	Nickname       string `json:"nickname"`         // 用户名
	UserLevel      int    `json:"user_level"`       // 用户等级
	Ip             string `json:"ip"`               // IP地址
	RoomID         int    `json:"room_id"`          // 直播间ID
	LastOnlineTime int64  `json:"last_online_time"` // 上次在线时间
}

func (o *OnlineDeviceRequest) UnmarshalBinary(data []byte) error {
	return json.Unmarshal(data, o)
}

type OnlineDevice struct {
	rds redis.UniversalClient
}

var (
	insOnline  *OnlineDevice
	onlineOnce sync.Once
)

func NewOnlineDevice() *OnlineDevice {
	onlineOnce.Do(func() {
		insOnline = &OnlineDevice{
			rds: global2.GetRds(),
		}
	})
	return insOnline
}

// ES Index 名称
func (o *OnlineDevice) genEsIndexName() string {
	return "online_device"
}

// key
func (o *OnlineDevice) genOnlineKey() string {
	return "online_device"
}

//func (o *OnlineDevice) genGuestOnlineKey() string {
//	return "online_device_guest"
//}

// WriteToRedis 写入Redis排队
func (o *OnlineDevice) WriteToRedis(req OnlineDeviceRequest) {
	bytes, _ := json.Marshal(req)
	o.rds.HSet(o.genOnlineKey(), req.UserID, bytesconv.BytesToString(bytes))
}

// DelOnlineDevice 删除在线设备
func (o *OnlineDevice) DelOnlineDevice(userId string) {
	o.rds.HDel(o.genOnlineKey(), userId)
}

func (o *OnlineDevice) GetAllOnlineDevice() (error, []interface{}) {
	var resp []interface{}
	var cursor uint64
	for {
		var err error
		var rs []string
		rs, cursor, err = o.rds.HScan(o.genOnlineKey(), cursor, "", 500).Result()
		if err != nil || len(rs) < 1 {
			global2.Log.Errorf("GetUserAllConversationCache err:%v", err)
			break
		}
		for k, v := range rs {
			if k%2 != 0 {
				onlineDevice := OnlineDeviceRequest{}
				err = json.Unmarshal([]byte(v), &onlineDevice)
				if err != nil {
					global2.Log.Errorf("err :%v", err)
					continue
				}
				resp = append(resp, onlineDevice)
			}
		}
		if cursor == 0 {
			break
		}
	}
	return nil, resp
}

// DelExpiredOnlineDevice 定时脚本执行
func (o *OnlineDevice) DelExpiredOnlineDevice() {
	err, rs := o.GetAllOnlineDevice()
	if err != nil {
		return
	}
	for _, v := range rs {
		resp, ok := v.(OnlineDeviceRequest)
		if !ok {
			global2.Log.Debugf("DelExpiredOnlineDevice:%+v,onlineDevice:%+v,ok:%v", v, resp, ok)
			continue
		}
		if resp.LastOnlineTime < time.Now().Unix()-7200 {
			o.DelOnlineDevice(resp.UserID)
		}
	}
}

// SyncToEs 同步到ES
func (o *OnlineDevice) SyncToEs() {
	err, result := o.GetAllOnlineDevice()
	if err != nil {
		return
	}
	records := make([]*elastic.BulkIndexRequest, 0)
	for _, v := range result {
		resp, ok := v.(OnlineDeviceRequest)
		if !ok {
			global2.Log.Debugf("DelExpiredOnlineDevice:%+v,onlineDevice:%+v,ok:%v", v, resp, ok)
			continue
		}
		resp.LastOnlineTime = time.Now().Unix()
		records = append(
			records,
			elastic.NewBulkIndexRequest().Index(o.genEsIndexName()).Doc(resp).Id(resp.UserID),
		)
	}

	//o.deleteEsOnlineData(o.genEsIndexName())
	if o.bulkToEs(records) {
		o.deleteEsOnlineData()
	}
}

func (o *OnlineDevice) deleteEsOnlineData() {
	query := elastic.NewRangeQuery("last_online_time").Lte(time.Now().Unix() - 25)
	_, err := global2.GetEs().DeleteByQuery(o.genEsIndexName()).Query(query).Do(context.Background())
	if err != nil {
		global2.Log.Errorf("deleteEsOnlineData err:%", err)
	}
}

func (o *OnlineDevice) bulkToEs(records []*elastic.BulkIndexRequest) bool {
	bulk := global2.GetEs().Bulk()
	for i := range records {
		bulk.Add(records[i])
	}
	if len(records) < 1 {
		return false
	}
	res, err := bulk.Do(context.Background())
	if err != nil {
		global2.Log.Errorf("exec bulk to es : %v", err)
		return false
	}
	for _, item := range res.Failed() {
		global2.Log.Errorf("write to es err: %s", item.Error.Reason)
	}
	return true
}

func (o *OnlineDevice) GetFromRedis(userId int) (device *OnlineDeviceRequest, exists bool) {
	device = &OnlineDeviceRequest{}
	err := o.rds.HGet(o.genOnlineKey(), strconv.Itoa(userId)).Scan(device)
	if err != nil && !errors.Is(err, redis.Nil) {
		global2.Log.Errorf("get from redis err: %s", err)
	}
	return
}

func (o *OnlineDevice) GetFromEs(userId string) (device *OnlineDeviceRequest, err error) {
	device = &OnlineDeviceRequest{}
	query := elastic.NewBoolQuery().Filter(elastic.NewTermsQuery("user_id", userId))
	res, err := global2.GetEs().Search().Index(o.genEsIndexName()).
		Query(query).
		Do(context.Background())
	if err != nil {
		return
	}
	global2.Log.Debugf("hit.Source res.Hits.TotalHits.Value:%v", res.Hits.TotalHits.Value)
	if res.Hits.TotalHits.Value > 0 {
		hit := res.Hits.Hits[0]
		global2.Log.Debugf("hit.Source:%v", hit.Source)
		err = json.Unmarshal(hit.Source, device)
		if err != nil {
			global2.Log.Errorf("GetFromEs Unmarshal err:%v", err)
			return nil, err
		}
	}
	return device, nil
}
